package com.hkbigdata.watermark;

import com.hkbigdata.bean.WaterSensor;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.JoinFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.Collector;

import java.time.Duration;

/**
 * @author liuanbo
 * @creat 2024-05-11-17:20
 * @see 2194550857@qq.com
 */
public class Flink06_ProcessFunction {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment().setParallelism(1);

        SingleOutputStreamOperator<WaterSensor> waterSensorSingleOutputStreamOperator = env.socketTextStream("localhost", 9999).flatMap(new FlatMapFunction<String, WaterSensor>() {
            @Override
            public void flatMap(String value, Collector<WaterSensor> out) throws Exception {
                String[] split = value.split(",");
                out.collect(new WaterSensor(split[0],
                        Long.valueOf(split[1]),
                        Double.valueOf(split[2])
                ));
            }
        });
        //1. 普通函数   new MapFunction<>() 单纯的只有一个处理对应逻辑方法
        //2.复函数 new RichMapFunction<>() 生命周期方法 和运行时上下文对象
        //3.process 生命周期方法 和运行时上下文对象 定时器和侧输出流和状态编程

        //TODO 1. ProcessFunction
/*        waterSensorSingleOutputStreamOperator.process(new ProcessFunction<WaterSensor, String>() {
            @Override
            public void processElement(WaterSensor value, Context ctx, Collector<String> out) throws Exception {
                out.collect("输出数据："+value);
            }

        });*/
        //TODO 2.KeyedProcessFunction
/*        waterSensorSingleOutputStreamOperator.keyBy(WaterSensor::getId)
                .process(new KeyedProcessFunction<String, WaterSensor, String>() {
                    @Override
                    public void processElement(WaterSensor value, Context ctx, Collector<String> out) throws Exception {
                        out.collect(ctx.getCurrentKey() + value);
                    }
                });*/
        //TODO 3.ProcessJoinFunction
        SingleOutputStreamOperator<WaterSensor> waterSensorSingleOutputStreamOperator1 = env.socketTextStream("localhost", 8888).flatMap(new FlatMapFunction<String, WaterSensor>() {
            @Override
            public void flatMap(String value, Collector<WaterSensor> out) throws Exception {
                String[] split = value.split(",");
                out.collect(new WaterSensor(split[0],
                        Long.valueOf(split[1]),
                        Double.valueOf(split[2])
                ));
            }
        });


        DataStream<WaterSensor> apply = waterSensorSingleOutputStreamOperator1.keyBy(WaterSensor::getId).assignTimestampsAndWatermarks(WatermarkStrategy.<WaterSensor>forBoundedOutOfOrderness(Duration.ofSeconds(3)).withTimestampAssigner(new SerializableTimestampAssigner<WaterSensor>() {
            @Override
            public long extractTimestamp(WaterSensor element, long recordTimestamp) {
                return element.getTs() * 1000L;
            }
        })).
                join(waterSensorSingleOutputStreamOperator1.keyBy(WaterSensor::getId))
                .where(new KeySelector<WaterSensor, String>() {
                    @Override
                    public String getKey(WaterSensor value) throws Exception {
                        return value.getId();
                    }
                }).equalTo(WaterSensor::getId).
                        window(TumblingEventTimeWindows.of(Time.seconds(5)))

                .allowedLateness(Time.seconds(2))

                .apply(new JoinFunction<WaterSensor, WaterSensor, WaterSensor>() {
                    @Override
                    public WaterSensor join(WaterSensor first, WaterSensor second) throws Exception {
                        return new WaterSensor(first.getId(), second.getTs(), first.getVc() + second.getVc());
                    }
                });

        //如果说抽象类，重写所有抽象方法
        //如果说接口，重写所有方法
//        apply.join()
        env.execute();
    }
}
